package com.pkk.spring.cloud.web.runner.controller.kafka.controller;

import com.pkk.spring.cloud.web.runner.controller.kafka.constant.KafkaConstant;
import com.pkk.spring.cloud.web.runner.controller.kafka.producer.SimpleKafkaProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @description: kafka的接口
 * @author: peikunkun
 * @create: 2021-05-24 16:58
 **/
@Slf4j
@RestController
@RequestMapping("kafka/simple")
public class KafkaSimpleController {

    @Autowired
    private SimpleKafkaProducer simpleKafkaProducer;
    @Autowired
    private KafkaTemplate kafkaTemplate;


    /**
     * 发送消息
     *
     * @param message
     * @return java.lang.String
     * @author peikunkun
     * @date 2021/5/24 17:04
     */
    @RequestMapping("send/{message}")
    public String send(@PathVariable(value = "message") Object message) {
        return simpleKafkaProducer.send(message);
    }


    /**
     * 发送kafka事物消息
     *
     * @param message
     * @return java.lang.String
     * @author kunkun.pei
     * @date 2021/6/9 18:55
     */
    @RequestMapping("sendt/{message}")
    public String sendt(@PathVariable(value = "message") Object message) {

        Object o = kafkaTemplate.executeInTransaction(op -> {
            op.send(KafkaConstant.SIMPLE_TOPIC, message);
            return "error".equals(message) ? 1 / 0 : "success";
        });

        String s = "err".equals(message) ? (1 / 0) + "2" : "success";
        return o.toString()+s;
    }


}
